package com.shujia.dwi

import com.shujia.utils.SparkTool
import com.shujia.utils.Config

/**
  * 抽取dpi,oidd ,ddr wcdr中的用户位置信息，构建位置融合表
  * 每天0点处理前一天的数据
  * 数据量将达，5点左右可以处理完
  *
  */
object MergeLocationJob extends SparkTool {
  /**
    * 子类实现此方法，实现具体的代码逻辑
    */
  override def run(args: Array[String]): Unit = {

    if (args.length == 0) {
      log.error("请指定时间参数：")
      //抛出异常
      throw new RuntimeException("请指定时间参数：day_id")
    }
    val day_id = args(0)
    log.info(s"当前处理的时间分区为：$day_id")

    val ddrPath = Config.get("ddr.path") + s"day_id=$day_id"
    val dpiPath = Config.get("dpi.path") + s"day_id=$day_id"
    val wcdrPath = Config.get("wcdr.path") + s"day_id=$day_id"
    val oiddPath = Config.get("oidd.path") + s"day_id=$day_id"


    //读取四类数据
    val ddr = sc.textFile(ddrPath)
    val dpi = sc.textFile(dpiPath)
    val wcdr = sc.textFile(wcdrPath)
    val oidd = sc.textFile(oiddPath)

    /**
      * 1、过滤脏数据
      * 2、抽取用户位置
      * 3、乒乓处理
      * 4、合并位置点
      */

    //合并四类数据
    val mergeLocation = ddr.union(dpi).union(wcdr).union(oidd)

    val outPutPath = Config.get("mergeLocation.path") + s"day_id=$day_id"

    //保存数据
    save(mergeLocation, outPutPath)

    /**
      * spark-submit --master yarn-client --class com.shujia.dwi.MergeLocationJob  --num-executors 4 dw-1.0.jar 20180503
      */
  }

  /**
    * 指定spark配置
    * conf.set(key,value)
    */
  override def init(): Unit = {
//    conf.setMaster("local[8]")
  }
}



